package com.gzx.plugin.tjzy.core.kafka;

import com.gzx.plugin.tjzy.modular.call.ht.service.GzxHtCallAsyncUpdateRecordService;
import com.gzx.plugin.tjzy.modular.call.receipt.service.GzxHtCallReceiptBusinessService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import vip.xiaonuo.common.kafka.KafkaTopicConstant;

import java.util.List;

@Slf4j
@Component
@RequiredArgsConstructor
public class EventKafkaConsumer {

    private final GzxHtCallAsyncUpdateRecordService htCallAsyncUpdateRecordService;

    private final GzxHtCallReceiptBusinessService htCallReceiptBusinessService;

    @KafkaListener(topics = "event-topic",
            groupId = "my-consumer-group", containerFactory = "batchFactory")
    public void listen(List<ConsumerRecord<String, String>> recordList, Acknowledgment ack) {
        log.info("从MQ中批量获取event：本批次处理event数量{}", recordList.size());
        log.info("Kafka收到消息：{}", recordList);
        try {
            htCallAsyncUpdateRecordService.asyncUpdateRecord(recordList);
            // 模拟延迟控制消费频率
            try {
                Thread.sleep(500);  // 延迟1秒，模拟消费频率控制
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            // 业务处理完毕再手动提交offset
            ack.acknowledge();
        } catch (Exception e) {
            e.printStackTrace();
            log.error("当前批次消息处理失败:{}", e.getMessage());
        }
    }

    @KafkaListener(topics = KafkaTopicConstant.CALL_RECEIPT_TOPIC,
            groupId = "call-receipt-group", containerFactory = "batchFactory")
    public void saveBatchBusiness(List<ConsumerRecord<String, String>> recordList, Acknowledgment ack) {
        log.info("从MQ中批量获取event：本批次处理event数量{}", recordList.size());
        log.info("Kafka收到消息：{}", recordList);
        try {
            htCallReceiptBusinessService.saveBatchBusiness(recordList);
            // 业务处理完毕再手动提交offset
            ack.acknowledge();
        } catch (Exception e) {
            log.error("当前批次消息处理失败:", e);
            throw e;
        }
    }
}